package org.example
import org.apache.spark.sql.SparkSession
object Object {
  def main(args: Array[String]): Unit = {
      val spark = SparkSession
        .builer()
        .master("local[*]")
        .appName("spark")
        .getOrCreate()
      val sc = spark.sparkContext

      val filePath = "src/main/resources/"
      val moviesRDD = sc.textFile(filePath + "movies.dat")
      val occupationsRDD = sc.textFile(filePath + "occupations.dat")
      val ratingsRDD = sc.textFile(filePath + "ratings.dat")
      val usersRDD = sc.textFile(filePath + "users.dat")
      println("总电影数：" + moviesRDD.count())
      println("总用户数：" + usersRDD.count())
      println("评分条数：" + ratingsRDD.count())
      println("职业数：" + occupationsRDD.count())


      val userRating = ratingsRDD.map(_.split("::"))
        .map(user => (user(1), user(0)))
        .filter(_._2.equals("2688"))
      println("用户ID为2688的影评数量为：" + userRating.count() + "以下为电影详情\n")

      val moveInfo = moviesRDD.map(_.split("::"))
        .map(movie => (movie(0), (movie(1)), (movie(2))))
        .join(userRating)
        .map(item => (item._1, item._2._2, item._2._1._1, item._2_1._2))
      movieInfo.foreach(println)


      //zuoye
      val rating = ratingsRDD.map(x => x.split("::")).map {
        x => {
          (x(0), x(1), x(2))
        }
      }.cache()
      println("平均得分最高的前 10 名的电影名称简单版")
      rating.map(x => (x._2, (x._3.toDouble, 1)))
        .reduceByKey((x, y) => {
          (x._1 + y._1, x._2 + y._2)
        })
        .map(x => (x._2._1 / x._2._2, x._1))
        .sortByKey(false)
        .take(5)
        .foreach(println)
      println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次 数))")
      val moviesInfo = moviesRDD.map(x => x.split("::"))
        .map(x => {
          (x(0), (x(1), x(2)))
        })
      val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
        .reduceByKey((x, y) => {
          (x._1 + y._1, x._2 + y._2)
        })
        .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2)))
      moviesInfo.join(ratingsInfo)
        .map(info => {
          (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
        }).sortByKey(false)
        .take(5)
        .foreach(println)
      println("观影人数最多的前 10 部电影")
      val watchViewsInfo = rating.map(x => {
          (x._2, 1)
        }).reduceByKey((x, y) => x + y)
        .map(x => (x._2, x._1))
        .sortByKey(false)
        .take(5)
      watchViewsInfo.foreach(println(_))
      println("===================>")
      rating.map(x => (x._2, 1))
        .reduceByKey((x, y) => {
          (x + y)
        })
        .map(x => (x._2, x._1))
        .sortByKey(false)
        .take(10)
        .foreach(println)
      println("详情的输出(  观影人数，电影编号)")
      moviesInfo.join(ratingsInfo).map(x => {
          (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
        }).sortByKey(false)
        .take(10)
        .foreach(println)

      println(movieInfo)

      //0411
      val movieStyle = moviesRDD.map(_.split("::"))
        .map(x => x(0), x(2))
        .flatMapValues(x => x.split("\\|"))
        .map(x => (x._2, 1))
        //在控制台打印科幻片的电影数量（Fantasy）
        .reduceByKey((x, y) => x + y)
      movieStyle.take(10).foreach(println)
      //正则表达式
      val pattern = Pattern.compile("(.*)(\\(\\d{4}\\))")

      val movieYear = moviesRDD.map(_.split("::"))
        .map(x => (x(1), 1))
        .map(x => {
          var name = ""
          var year = ""
          val matcher = pattern.matcher(x._1)
          if (matcher.find()) {
            name = matcher.group(1)
            year = matcher.group(2)
            year = year.substring(1, year.length - 1)
          }
          if (year == "") {
            (-1, 1) //标识码
          } else {
            (year.toInt, 1)
          }
        })
        .reduceByKey((x, y) => x + y)


        //求2000年度电影生产数量
        .sortByKey()
        .collect()
      movieYear.foreach(println)


      sc.stop()
    }




}
